package com.tc.juc.pack08;

import lombok.extern.slf4j.Slf4j;

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @Auther: tianchao
 * @Date: 2021/10/7 15:37
 * @Description:
 */
@Slf4j
public class TestCustomerPool {
    public static void main(String[] args) {
        MyThreadPool myThreadPool = new MyThreadPool(1, TimeUnit.MILLISECONDS, 3000, 1, new MyRejectPolicy<Runnable>() {
            @Override
            public void reject(MyBlockingQueue<Runnable> blockingQueue, Runnable task) {
                //1 死等
                // blockingQueue.put(task);
                //2 超时等待 超时丢弃
                //boolean offer = blockingQueue.offer(task, 500, TimeUnit.MILLISECONDS);
                //log.info("加入成功了吗: {}",offer);
                //3 直接任务执行
                //log.info("放弃任务: {}",task);
                //4 抛出异常
                throw new RuntimeException("线程已满,放弃任务"+ task);
                //5 让调用者自己执行任务
//                new Thread(()->{
//                    task.run();
//                },"myself线程").start();

            }
        });
        for (int i = 0; i < 15; i++) {
            final int temp = i;
            myThreadPool.execute(()->{
                try {
                    Thread.sleep(10000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                log.info("执行的任务内容,第几个任务：{}",temp);
            });
        }
        try {
            Thread.sleep(5000);
            log.info("休眠结束===========================");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
//        for (int i = 0; i < 5; i++) {
//            final int temp = i;
//            myThreadPool.execute(()->{
//                log.info("执行的任务内容,第几个任务：{}",temp);
//            });
//        }
    }

}
@Slf4j
class MyThreadPool{
    //任务队列
    private MyBlockingQueue<Runnable> taskQueue;
    //线程集合
    private HashSet<Worker> workers = new HashSet();
    //核心线程数
    private int coreSize;
    //获取任务的超时时间
    private TimeUnit timeUnit;
    private long timeOut;
    private MyRejectPolicy<Runnable> rejectPolicy;

    public MyThreadPool(int coreSize, TimeUnit timeUnit, long timeOut,int queueCapacity,MyRejectPolicy<Runnable> rejectPolicy) {
        this.coreSize = coreSize;
        this.timeUnit = timeUnit;
        this.timeOut = timeOut;
        this.taskQueue = new MyBlockingQueue<>(queueCapacity);
        this.rejectPolicy = rejectPolicy;

    }
    //执行任务
    public void execute(Runnable task){
        //当任务数没有超过核心线程数 coreSize,直接交给worker对象执行
        //当任务数超过核心线程数 coreSize,放入任务队列缓存
        synchronized (workers){
            if (workers.size()<coreSize){
                Worker worker = new Worker(task);
                log.info("新增worker: {}, 任务: {}",worker,task);
                workers.add(worker);
                worker.start();
            }else{
                log.info("加入队列 task: {}",task);
                taskQueue.tryPut(rejectPolicy,task);
                // taskQueue.put(task);
                //1 死等
                //2 超时等待
                //3 直接任务执行
                //4 抛出异常
                //5 让调用者自己执行任务
            }
        }
    }

    class Worker extends Thread{
        Runnable task;

        public Worker(Runnable task) {
            log.info("创建worker task内容: {}", task);
            this.task = task;
        }

        @Override
        public void run() {
            // 1任务不是空 执行任务
            // 2 task执行完毕,从任务队列获取任务并执行
            while (task!=null|| (task = taskQueue.poll(timeOut, timeUnit))!=null){
                try {
                    log.info("任务正在执行...:   {}", task);
                    task.run();
                }catch (Exception e){
                    e.printStackTrace();
                }finally {
                    task = null;
                }
            }
            synchronized (workers){
                log.info("worker被移除: {}",this);
                workers.remove(this);
            }
        }
    }
}
@FunctionalInterface
interface MyRejectPolicy<T>{
    void reject(MyBlockingQueue<T> blockingQueue, T task);
}

@Slf4j
class MyBlockingQueue<T>{
    //任务队列
    private Deque<T> queue = new ArrayDeque<>();
    //锁
    private ReentrantLock lock = new ReentrantLock();
    //生产者条件变量
    private Condition fullWaitSet = lock.newCondition();
    //消费者条件变量
    private Condition emptyWaitSet = lock.newCondition();

    private int capacity;

    public MyBlockingQueue(int capacity) {
        this.capacity = capacity;
    }

    // 带超时的获取方法
    public T poll(long timeOut, TimeUnit timeUnit){
        lock.lock();
        try {
            long nacos = timeUnit.toNanos(timeOut);

            while (queue.isEmpty()){
                try {
                    //返回了剩余时间
                    if (nacos<=0){
                        return null;
                    }
                    nacos = emptyWaitSet.awaitNanos(nacos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            T element = queue.removeFirst();
            fullWaitSet.signal();
            return element;
        }finally {
            lock.unlock();
        }
    }

    //阻塞获取
    public T take(){
        lock.lock();
        try {
            while (queue.isEmpty()){
                try {
                    emptyWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            T element = queue.removeFirst();
            fullWaitSet.signal();
            return element;
        }finally {
            lock.unlock();
        }
    }

    public void put(T task){
        lock.lock();
        try {
            while (queue.size() == capacity){
                try {
                    log.info("等待加入任务队列");
                    fullWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            queue.addLast(task);
            emptyWaitSet.signal();
        }finally {
            lock.unlock();
        }
    }

    /**
     * put的意思
     * @return
     */
    public boolean offer(T task,long timeOut,TimeUnit timeUnit){
        lock.lock();
        try {
            long nacos = timeUnit.toNanos(timeOut);
            while (queue.size() == capacity){
                try {
                    log.info("等待加入任务队列");
                    if (nacos<=0){
                        return false;
                    }
                    nacos = fullWaitSet.awaitNanos(nacos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                    return false;
                }
            }
            queue.addLast(task);
            emptyWaitSet.signal();
            return true;
        }finally {
            lock.unlock();
        }
    }

    //获取大小

    public int size(){
        lock.lock();
        try {
            return queue.size();
        }finally {
            lock.unlock();
        }
    }

    public void tryPut(MyRejectPolicy<T> rejectPolicy, T task) {
        lock.lock();
        try {
            //队列已经满了
            if (queue.size() == capacity){
                rejectPolicy.reject(this, task);
            }else{
                log.info("加入任务队列");
                queue.addLast(task);
                emptyWaitSet.signal();
            }
        }finally {
            lock.unlock();
        }
    }
}
